package com.atguigu.day04;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

public class Example6 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> "click")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<ClickEvent, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<ClickEvent> elements, Collector<String> collector) throws Exception {
                        // key: username
                        // value: 用户访问次数
                        var map = new HashMap<String, Integer>();

                        for (var e : elements) {
                            if (!map.containsKey(e.username)) {
                                map.put(e.username, 1);
                            } else {
                                map.put(e.username, map.get(e.username) + 1);
                            }
                        }

                        var arrayList = new ArrayList<Tuple2<String, Integer>>();
                        for (var username : map.keySet()) {
                            arrayList.add(Tuple2.of(username, map.get(username)));
                        }

                        arrayList.sort(new Comparator<Tuple2<String, Integer>>() {
                            @Override
                            public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
                                return o2.f1 - o1.f1;
                            }
                        });

                        collector.collect("窗口" + new Timestamp(context.window().getStart()) + "~" +
                                "" + new Timestamp(context.window().getEnd()) + "里面访问次数最多的用户是：" +
                                "" + arrayList.get(0).f0 + ", 浏览次数是：" + arrayList.get(0).f1);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
